package org.example.data_stream;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * 复函数
 * @author shenguangyang
 */
public class E09_TransformRichFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user02", "./cart", 2000L));
            add(new Event("user01", "./home1", 1000L));
            add(new Event("user02", "./prod?id=10`", 3800L));
            add(new Event("user02", "./prod?id=2`", 3000L));
            add(new Event("user01", "./prod?id=15`", 4000L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);

        streamSource.map(new MyRichMapper()).print();
        env.execute();
    }

    private static class MyRichMapper extends RichMapFunction<Event, Integer> {
        @Override
        public Integer map(Event event) throws Exception {
            return event.getUrl().length();
        }

        /**
         * 任务实例创建的时候会调用
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println("open 生命周期被调用: " + getRuntimeContext().getIndexOfThisSubtask() + " 号任务启动了");
        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println("close 生命周期被调用: " + getRuntimeContext().getIndexOfThisSubtask() + " 号任务结束了");
        }
    }
}
